package com.afdor.rws.event.driven.channel.aliyun.mns.producer.topic;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.core.exception.BusinessException;
import com.afdor.rws.event.driven.api.channel.producer.EventProducer;
import com.afdor.rws.event.driven.channel.aliyun.mns.config.AliyunMnsEventConfig;
import com.afdor.rws.event.driven.channel.aliyun.mns.utils.AliyunMnsUtils;
import com.afdor.rws.event.driven.config.EventProducerConfig;
import com.afdor.rws.event.driven.entities.Event;
import com.afdor.rws.event.driven.support.channel.producer.AbstractEventProducer;
import com.afdor.rws.event.driven.support.holder.CallbackEvent;
import com.afdor.rws.event.driven.support.holder.EventResult;
import com.aliyun.mns.client.CloudAccount;
import com.aliyun.mns.client.CloudTopic;
import com.aliyun.mns.model.RawTopicMessage;
import lombok.Getter;
import lombok.NoArgsConstructor;

import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.Optional;

/**
 * 阿里云MNS(消息服务)基于topic(主题)实现的生成者
 *
 * @author 悭梵
 * @date Created in 2018-12-13 14:43
 */
@NoArgsConstructor
@Extension(value = "aliyunMnsTopic", order = Extension.NORMAL_ORDER - 50, scope = Extension.Scope.PROTOTYPE, interfaceClass = EventProducer.class)
public class AliyunMnsEventProducer extends AbstractEventProducer<EventProducerConfig> {

    @Getter
    private CloudAccount cloudAccount;
    @Getter
    private CloudTopic topic;

    @Override
    protected void doInit() {
        if (this.cloudAccount != null) {
            throw new RuntimeException("已执行初始化操作，请不要再重复执行初始化操作了。");
        }

        AliyunMnsEventConfig aliyunConfig = (AliyunMnsEventConfig) getConfig().getEventConfig();
        this.cloudAccount = new CloudAccount(aliyunConfig.getSecretId(), aliyunConfig.getSecretKey(), aliyunConfig.getEndpoint());

        String topicName = AliyunMnsUtils.getTopicName(getEventSourceName());
        this.topic = this.cloudAccount.getMNSClient().getTopicRef(topicName);
        this.topic.create();
    }

    @Override
    public void sendAsyncEvent(Event event) {
        if (Objects.isNull(event.getSendType())) {
            event.setSendType(Event.SendType.ASYNC);
        }
        event.setStatus(Event.Status.SEND);
        RawTopicMessage message = new RawTopicMessage();
        message.setMessageBody((String) getSerialization().serialize(event));
        getTopic().publishMessage(message);
    }

    @Override
    public void batchSendAsyncEvent(List<Event> eventList) {
        Optional.ofNullable(eventList)
                .orElse(Collections.emptyList())
                .forEach(event -> {
                    event.setSendType(Event.SendType.ASYNC_BATCH);
                    sendAsyncEvent(event);
                });
    }

    @Override
    public void asyncSendEvent(CallbackEvent callbackEvent) {
        throw new BusinessException("阿里云MNS(消息服务)不支持同步回调消息发送");
    }

    @Override
    public <V> EventResult<V> sendEvent(Event event) {
        throw new BusinessException("阿里云MNS(消息服务)不支持同步回调消息发送");
    }

    @Override
    protected void doDestroy() {
        if (this.topic != null) {
            this.topic = null;
        }
        if (this.cloudAccount != null) {
            this.cloudAccount.getMNSClient().close();
            this.cloudAccount = null;
        }
    }
}
